package com.bitdf.txing.search.scheduled;

import com.alibaba.fastjson.JSON;
import com.bitdf.txing.common.utils.R;
import com.bitdf.txing.search.config.TxElasticConfig;
import com.bitdf.txing.search.constant.EsConstant;
import com.bitdf.txing.search.feign.EsGroupFeignService;
import com.bitdf.txing.search.to.GroupInfoTo;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.redisson.api.RBloomFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


/**
 * 定时任务
 * 1、@EnableScheduling 开启定时任务
 * 2、@Scheduled开启一个定时任务
 * <p>
 * 异步任务
 * 1、@EnableAsync:开启异步任务
 * 2、@Async：给希望异步执行的方法标注
 */

@Slf4j
@Component
// @EnableAsync
@EnableScheduling
public class EsScheduled {

    @Autowired
    RestHighLevelClient restHighLevelClient;
    @Autowired
    EsGroupFeignService groupFeignService;
    @Autowired
    RBloomFilter bloomFilter;

    /**
     * 1、在Spring中表达式是6位组成，不允许第七位的年份
     * 2、在周几的的位置,1-7代表周一到周日
     * 3、定时任务不该阻塞。默认是阻塞的
     * 1）、可以让业务以异步的方式，自己提交到线程池
     * CompletableFuture.runAsync(() -> {
     * },execute);
     * <p>
     * 2）、支持定时任务线程池；设置 TaskSchedulingProperties
     * spring.task.scheduling.pool.size: 5
     * <p>
     * 3）、让定时任务异步执行
     * 异步任务
     * <p>
     * 解决：使用异步任务 + 定时任务来完成定时任务不阻塞的功能
     */

    /**
     * 凌晨三点 定时更新数据库（把es中的组团数据同步到数据库）
     *
     * @throws Exception
     */
    @Async
    @Scheduled(cron = "0 0 3 * * *")
    public void updateData() throws Exception {
        log.info("[定时任务开始执行] ==> es 同步到 mysql");
        //查询出es中的所有组团数据
        SearchRequest searchRequest = new SearchRequest(EsConstant.GROUP_INDEX);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.termQuery("modifyFlag", 1)).size(Integer.MAX_VALUE);
        sourceBuilder.trackTotalHits(true);
        searchRequest.source(sourceBuilder);

        List<GroupInfoTo> list = new ArrayList<>();
        //若出现异常，则重试，最多重试三次
        int i = 3;
        while (i > 0) {
            try {
                SearchResponse response = restHighLevelClient.search(searchRequest, TxElasticConfig.COMMON_OPTIONS);
                for (SearchHit hit : response.getHits()) {
                    if (hit != null) {
                        GroupInfoTo groupInfoTo = JSON.parseObject(hit.getSourceAsString(), GroupInfoTo.class);
                        list.add(groupInfoTo);
                    }
                }
                //远程调用tx-group服务更新数据库
                R r = groupFeignService.syncDatabase(list);
                if (r.getCode() != 200) {
                    throw new RuntimeException();
                }
                break;
            } catch (Exception e) {
                i--;
                Thread.sleep(1000 * 60);
            }
        }
        //输出日志
        if (i == 0) {
            log.error("es组团数据同步到数据库时出现了异常");
        } else {
            // 更新成功后 把各组团修改标志位重置为0
            BulkRequest bulkRequest = new BulkRequest();
            for (GroupInfoTo groupInfoTo : list) {
                UpdateRequest updateRequest = new UpdateRequest(EsConstant.GROUP_INDEX, groupInfoTo.getId().toString());
                updateRequest.doc(XContentType.JSON, "modifyFlag", 0);
                // 添加批量操作
                bulkRequest.add(updateRequest);
            }
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, TxElasticConfig.COMMON_OPTIONS);
            log.info("es组团数据成功同步到数据库");
        }
    }


    /**
     * 定期更新布隆过滤器
     */
    @Async
    @Scheduled(cron = "0 0 3 * * *")
    public void updateRBloomFilter() throws IOException {
        log.info("定时任务触发，更新布隆过滤器");
        //查询出es中的所有组团数据ids
        SearchRequest searchRequest = new SearchRequest(EsConstant.GROUP_INDEX);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchAllQuery()).size(Integer.MAX_VALUE);
        sourceBuilder.trackTotalHits(true);
        searchRequest.source(sourceBuilder);
        // 指定返回字段
        String[] includeFields = new String[]{"id"};
        String[] exclude = new String[]{""};
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includeFields, exclude);
        sourceBuilder.fetchSource(fetchSourceContext);
        // 查询
        SearchResponse response = restHighLevelClient.search(searchRequest, TxElasticConfig.COMMON_OPTIONS);
        // 重置布隆过滤器
        long value = response.getHits().getTotalHits().value;
        bloomFilter.delete();
        // 预期容量设置比当前大2000，可根据组团数增长率作调整
        bloomFilter.tryInit(value + 2000, 0.01);
        // 重新添加到布隆过滤器
        for (SearchHit hit : response.getHits()) {
            if (hit != null) {
                GroupInfoTo groupInfoTo = JSON.parseObject(hit.getSourceAsString(), GroupInfoTo.class);
                bloomFilter.add(groupInfoTo.getId());
            }
        }
        log.info("布隆过滤器 更新完毕， 当前组团数： {}", bloomFilter.count());
    }

}
